package com.wei.inputformat;

import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.BytesWritable;
import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.mapreduce.InputSplit;
import org.apache.hadoop.mapreduce.RecordReader;
import org.apache.hadoop.mapreduce.TaskAttemptContext;
import org.apache.hadoop.mapreduce.lib.input.FileSplit;

public class WholeFileRecordReader extends RecordReader<NullWritable, BytesWritable>{
	private FileSplit fileSplit;
	private Configuration configuration;
	private BytesWritable value=new BytesWritable();
	private boolean processed=false;

	@Override
	public void initialize(InputSplit split, TaskAttemptContext context)
			throws IOException, InterruptedException {
		this.fileSplit=(FileSplit) split;
		this.configuration=context.getConfiguration();
	}

	@Override
	public boolean nextKeyValue() throws IOException, InterruptedException {
		if (!processed) {
			long length = fileSplit.getLength();
			byte[] context=new byte[(int) length];
			Path path = fileSplit.getPath();
			FileSystem fileSystem = path.getFileSystem(configuration);
			FSDataInputStream open = null;
			try {
				open=fileSystem.open(path);
				IOUtils.readFully(open, context, 0, context.length);
				value.set(context, 0, context.length);
			} catch (Exception e) {
				IOUtils.closeStream(open);
			}
			processed=true;
			return true;
		}
		return false;
	}

	@Override
	public NullWritable getCurrentKey() throws IOException, InterruptedException {
		return NullWritable.get();
	}

	@Override
	public BytesWritable getCurrentValue() throws IOException, InterruptedException {
		return value;
	}

	@Override
	public float getProgress() throws IOException, InterruptedException {
		return processed ? 1F : 0;
	}

	@Override
	public void close() throws IOException {
		
	}

}
